package function;

import bean.socketBean;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import java.util.*;


public class TopNProcesstionFuncton extends ProcessAllWindowFunction<socketBean,socketBean, TimeWindow> {

    @Override
    public void process(Context context, Iterable<socketBean> elements, Collector<socketBean> out) throws Exception {
        Map<String,socketBean> m = new HashMap<String,socketBean>();
        List<socketBean> l = new ArrayList<socketBean>();
        Iterator<socketBean> iterator = elements.iterator();
        while (iterator.hasNext()){
            socketBean socketBean = iterator.next();
            m.put(socketBean.getId(),socketBean);
            l.add(socketBean);
        }
        l.sort(Comparator.comparing(socketBean::getMoney).reversed());
        for (int i=0;i<l.size();i++){
            if(i<3){
                out.collect(m.get(l.get(i).getId()));
            }
        }
    }
}
